package com.shujia.sql

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._

object Demo01CreateTableEnv {
  def main(args: Array[String]): Unit = {
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() // 使用Blink的SQL解析器
      .inStreamingMode() // 开启流处理模式
      .build()
    // 构建Table Env
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    // 基于stus Topic数据 统计每个班级的人数

    // 构建Source表 动态表
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE stus (
          | id BIGINT,
          | name STRING,
          | age BIGINT,
          | gender STRING,
          | clazz STRING
          |) WITH (
          | 'connector' = 'kafka',
          | 'topic' = 'stus',
          | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
          | 'properties.group.id' = 'test1',
          | 'format' = 'csv',
          | 'csv.field-delimiter' = ',',
          | 'scan.startup.mode' = 'earliest-offset'
          |)
          |""".stripMargin)



    //    // Transformation
    //    val clazzCntTable: Table = bsTableEnv
    //      .sqlQuery(
    //        """
    //          |select clazz,count(1) as cnt from stus group by clazz
    //          |""".stripMargin)

    /**
     * 打印Table
     * 需要先将Table转换成DS再打印
     * 结果数据需要更新：toRetractStream
     * 结果数据不需要更新：toAppendStream
     *
     * 最终输出的结果的格式：(Boolean, (String, Long))
     * 7> (false,(理科三班,67))
     * 7> (true,(理科三班,68))
     * 布尔值表示数据是否有更新
     * 当为true时表示数据有更新
     * 当为false时表示数据可以和之前的数据抵消
     */
    //    val clazzCntDS: DataStream[(Boolean, (String, Long))] = clazzCntTable.toRetractStream[(String, Long)]
    //    clazzCntDS.print()
    //    bsEnv.execute()


    // 构建Sink表 动态表
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE print_table (
          | clazz String,
          | cnt BIGINT
          |) WITH (
          | 'connector' = 'print'
          |)
          |""".stripMargin)

    // 创建Sink MySQL的表
    bsTableEnv
      // 当在处理数据时 需要更新最后的结果 需要在sink table中指定PRIMARY KEY
      // 并且MySQL中对应的表需要设置相同的主键 这样最终的结果就会按照PRIMARY KEY自动更新
      .executeSql(
        """
          |CREATE TABLE clazz_cnt (
          |  clazz STRING,
          |  cnt BIGINT,
          |  PRIMARY KEY (clazz) NOT ENFORCED
          |) WITH (
          |   'connector' = 'jdbc',
          |   'url' = 'jdbc:mysql://master:3306/student',
          |   'table-name' = 'clazz_cnt',
          |   'username' = 'root',
          |   'password' = '123456'
          |)
          |""".stripMargin)

    // 进行查询并将查询结果写入sink表
    bsTableEnv
      .executeSql(
        """
          |insert into clazz_cnt
          |select clazz,count(1) as cnt from stus group by clazz
          |""".stripMargin)

  }

}
